/**
 * Copyright © 2011 Creativor Studio.
 * About license information, please see LICENSE.txt.
 */
package org.creativor.rayson.transport.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.creativor.rayson.transport.api.Connection;
import org.creativor.rayson.transport.common.ConnectionProtocol;
import org.creativor.rayson.transport.common.ConnectionState;
import org.creativor.rayson.transport.common.Packet;
import org.creativor.rayson.transport.common.PacketConnection;
import org.creativor.rayson.transport.common.PacketCounter;
import org.creativor.rayson.transport.common.PacketException;
import org.creativor.rayson.transport.common.PacketReader;
import org.creativor.rayson.transport.common.PacketWithType;
import org.creativor.rayson.transport.common.ResponseType;
import org.creativor.rayson.util.Log;

// TODO: Auto-generated Javadoc
/**
 * An {@link Connection} which transport RPC call in server side.
 * 
 * @author Nick Zhang
 */
class RpcConnection extends PacketConnection
{
	
	private class PacketWriter
	{
		/**
		 * Last packet to write.
		 */
		private PacketWithType lastPacketCarrier;
		private ByteBuffer writeDataBuffer;
		
		public void write() throws IOException
		{
			if (this.lastPacketCarrier == null)
			{
				packetsLock.lock();
				try
				{
					this.lastPacketCarrier = sendPackets.remove();
				} finally
				{
					packetsLock.unlock();
				}
				byte code = this.lastPacketCarrier.getType();
				int dataLength = this.lastPacketCarrier.getPacket().getDataLength();
				byte[] data = this.lastPacketCarrier.getPacket().getData();
				// short checksum = CRC16.compute(code, dataLength, data);
				this.writeDataBuffer =
						ByteBuffer.allocate(this.lastPacketCarrier.getPacket().getDataLength()
								+ ConnectionProtocol.PACKET_HEADER_LENGTH);
				this.writeDataBuffer.put(code);
				this.writeDataBuffer.putShort((short) dataLength);
				// this.writeDataBuffer.putShort(checksum);
				this.writeDataBuffer.put(data);
				this.writeDataBuffer.flip();
			}
			
			RpcConnection.this.socketChannel.write(this.writeDataBuffer);
			if (!this.writeDataBuffer.hasRemaining())
			{
				// LOGGER.info("Write packet " +
				// this.lastPacketCarrier.toString()
				// + " out!");
				touch();
				packetCounter.writeOne();
				this.lastPacketCarrier = null;
				if (gotErrorPacket.get()) return;
				// we need to unregister the write event.
				packetsLock.lock();
				try
				{
					if (sendPackets.isEmpty()) selectionKey.interestOps(SelectionKey.OP_READ);
				} finally
				{
					packetsLock.unlock();
				}
			}
		}
	}
	
	private static Logger LOGGER = Log.getLogger();
	private byte version = -1;
	
	private AtomicBoolean closed;
	
	private ByteBuffer connectHeaderBuffer;
	private ConnectionState connectionState;
	private ByteBuffer connectResponseBuffer;
	
	private ByteBuffer errorResponseBuffer;
	
	private AtomicBoolean gotErrorPacket;
	private long id;
	private PacketCounter packetCounter;
	private PacketManager packetManager;
	
	private PacketReader packetReader;
	
	private PacketWriter packetWriter;
	
	private boolean readedConnectHeader = false;
	private SelectionKey selectionKey;
	private Queue<PacketWithType> sendPackets;
	private Lock packetsLock;
	private SocketChannel socketChannel;
	private boolean wroteConnectCode = false;
	private String addressInfo;
	private InetSocketAddress remoteAddr;
	/**
	 * Server configuration setting.
	 */
	private ServerConfig serverConfig;
	
	/**
	 * Instantiates a new rpc connection.
	 * 
	 * @param id
	 *            the id
	 * @param clientChannel
	 *            the client channel
	 * @param packetManager
	 *            the packet manager
	 * @param selectionKey
	 *            the selection key
	 * @param serverConfig
	 *            Server configuration setting.
	 */
	RpcConnection(long id, SocketChannel clientChannel, PacketManager packetManager,
			SelectionKey selectionKey, ServerConfig serverConfig)
	{
		this.id = id;
		// header length-protocol
		connectHeaderBuffer = ByteBuffer.allocate(ConnectionProtocol.HEADER_LENGTH - 1);
		connectResponseBuffer = ByteBuffer.allocate(ConnectionProtocol.RESPONSE_LENGTH);
		this.packetManager = packetManager;
		this.serverConfig = serverConfig;
		closed = new AtomicBoolean(false);
		sendPackets = new LinkedList<PacketWithType>();
		packetsLock = new ReentrantLock();
		packetCounter = new PacketCounter();
		gotErrorPacket = new AtomicBoolean(false);
		this.selectionKey = selectionKey;
		this.socketChannel = clientChannel;
		this.addressInfo = this.socketChannel.socket().toString();
		packetWriter = new PacketWriter();
		setConnectionState(ConnectionState.OK);
		packetReader = new ServerPacketReader(this.socketChannel);
	}
	
	private void addReceivedPacket(Packet packet) throws IOException
	{
		// LOGGER.info("Read packet:" + packet.toString());
		this.touch();
		this.packetCounter.readOne();
		// add packet to manager
		this.packetManager.addReceived(this, packet);
	}
	
	/**
	 * Adds the send packet.
	 * 
	 * @param packet
	 *            the packet
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	@Override
	public void addSendPacket(Packet packet) throws IOException
	{
		packetsLock.lock();
		try
		{
			this.sendPackets.add(new PacketWithType(ResponseType.OK.getType(), packet));
			if (this.sendPackets.size() == 1)
			{
				this.selectionKey.interestOps(this.selectionKey.interestOps()
						| SelectionKey.OP_WRITE);
				this.selectionKey.selector().wakeup();
			}
		} finally
		{
			packetsLock.unlock();
		}
	}
	
	/**
	 * Close.
	 * 
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	@Override
	public synchronized void close() throws IOException
	{
		if (!closed.compareAndSet(false, true)) return;
		try
		{
			// close the socket input output first.
			this.socketChannel.socket().shutdownInput();
			this.socketChannel.socket().shutdownOutput();
		} finally
		{
			this.socketChannel.socket().close();
		}
	}
	
	/**
	 * Gets the id.
	 * 
	 * @return the id
	 */
	@Override
	public long getId()
	{
		return id;
	}
	
	/**
	 * Gets the version.
	 * 
	 * @return the version
	 */
	@Override
	public byte getVersion()
	{
		return version;
	}
	
	private void gotErrorPacket(ResponseType responseType) throws IOException
	{
		gotErrorPacket.set(true);
		// Initialize error response buffer.
		byte type = responseType.getType();
		short dataLength = 0;
		byte[] data = new byte[dataLength];
		// short checksum = CRC16.compute(type, dataLength, data);
		errorResponseBuffer = ByteBuffer.allocate(ConnectionProtocol.PACKET_HEADER_LENGTH);
		errorResponseBuffer.put(type);
		errorResponseBuffer.putShort(dataLength);
		// errorResponseBuffer.putShort(checksum);
		errorResponseBuffer.flip();
		// do not accept any packet any more.
		this.selectionKey.interestOps(SelectionKey.OP_WRITE);
		this.socketChannel.socket().shutdownInput();
	}
	
	/**
	 * Inits the.
	 * 
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	@Override
	public void init() throws IOException
	{
		this.socketChannel.read(connectHeaderBuffer);
		if (!connectHeaderBuffer.hasRemaining())
		{
			connectHeaderBuffer.flip();
			version = connectHeaderBuffer.get();
			if (!isSupportedVersion(version))
				setConnectionState(ConnectionState.UNSUPPORTED_VERSION);
			this.selectionKey.interestOps(SelectionKey.OP_WRITE | SelectionKey.OP_READ);
			readedConnectHeader = true;
			this.remoteAddr =
					new InetSocketAddress(this.socketChannel.socket().getInetAddress()
							.getHostAddress(), this.socketChannel.socket().getPort());
		}
	}
	
	/**
	 * Checks if is supported version.
	 * 
	 * @param testVersion
	 *            the version
	 * @return true, if is supported version
	 */
	boolean isSupportedVersion(short testVersion)
	{
		return true;
	}
	
	/**
	 * Gets the timeout interval. If pending packet count of this connection is
	 * >0, then the time out interval will equals to
	 * {@link ServerConfig#pendingConnectionTimeout()} . Or else, it return
	 * {@link ConnectionProtocol#RPC_TIME_OUT_INTERVAL}.
	 */
	@Override
	protected long getTimeoutInterval()
	{
		if (this.pendingPacketCount() > 0) return serverConfig.pendingConnectionTimeout();
		return ConnectionProtocol.RPC_TIME_OUT_INTERVAL;
	}
	
	/**
	 * Pending packet count.
	 * 
	 * @return the int
	 */
	@Override
	public int pendingPacketCount()
	{
		return (int) this.packetCounter.readPendingCount();
	}
	
	/**
	 * Read.
	 * 
	 * @return the int
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	@Override
	public int read() throws IOException
	{
		
		// if (readedConnectHeader)
		{
			// add threshold control here.
			if (this.pendingPacketCount() >= ConnectionProtocol.MAX_PENDING_PACKETS) return 0;
			int readCount = 0;
			try
			{
				readCount = packetReader.read();
			} catch (PacketException e)
			{
				LOGGER.log(Level.SEVERE, "Read packet error: " + e.getMessage());
				gotErrorPacket(ResponseType.UNKNOW_REQUEST_Type);
				return 0;
			}
			if (packetReader.isReady())
			{
				addReceivedPacket(packetReader.takeLastWithType().getPacket());
			}
			return readCount;
		}
		// else
		// {
		// init();
		// return 0;
		// }
	}
	
	private void setConnectionState(ConnectionState connectionCode)
	{
		this.connectionState = connectionCode;
		this.connectResponseBuffer.clear();
		this.connectResponseBuffer.put(connectionCode.getState());
		this.connectResponseBuffer.clear();
	}
	
	private boolean shouldWriteErrorResponse()
	{
		return gotErrorPacket.get() && this.sendPackets.isEmpty();
	}
	
	/**
	 * Gets the remote addr.
	 * 
	 * @return the remote addr
	 */
	public InetSocketAddress getRemoteAddr()
	{
		return remoteAddr;
	}
	
	/**
	 * To string.
	 * 
	 * @return the string
	 */
	@Override
	public String toString()
	{
		StringBuffer sb = new StringBuffer();
		sb.append("{");
		sb.append("id: ");
		sb.append(this.id);
		sb.append(", protocol: ");
		sb.append(this.getProtocol());
		sb.append(", version: ");
		sb.append(this.getVersion());
		sb.append(", last contact: ");
		sb.append(getLastContactTime());
		sb.append(", packet counter: ");
		sb.append(this.packetCounter.toString());
		sb.append(", pending packets: ");
		sb.append(this.pendingPacketCount());
		sb.append(", address: ");
		sb.append(this.addressInfo);
		sb.append("}");
		return sb.toString();
	}
	
	/**
	 * Write.
	 * 
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	@Override
	public void write() throws IOException
	{
		if (wroteConnectCode)
		{
			if (shouldWriteErrorResponse())
			{
				writeErrorResponse();
			} else
			{
				packetWriter.write();
			}
			
		} else
		{
			// write connection response code.
			this.socketChannel.write(connectResponseBuffer);
			if (!this.connectResponseBuffer.hasRemaining())
			{
				// un-register write event. At that time, no packet has received
				// yet.
				this.selectionKey.interestOps(SelectionKey.OP_READ);
				wroteConnectCode = true;
				// try close this connection itself.
				switch (connectionState)
				{
					case SERVICE_UNAVALIABLE:
						try
						{
							this.close();
						} catch (IOException e)
						{
							// Ignore.
						}
						break;
					case UNSUPPORTED_VERSION:
						throw new IOException("Wrong version");
						
					default:
						break;
				}
				
			}
		}
	}
	
	private void writeErrorResponse() throws IOException
	{
		this.socketChannel.write(errorResponseBuffer);
		if (!errorResponseBuffer.hasRemaining())
		{
			// unregister the write event, and make this connection can read
			// EOF.
			this.selectionKey.interestOps(SelectionKey.OP_READ);
			this.socketChannel.socket().shutdownOutput();
		}
	}
}
